package belf.migrate.engine.job.structure;

import belf.migrate.api.common.JobMode;
import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.exception.TableNotExistException;
import belf.migrate.api.job.Job;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.table.schema.CatalogTable;
import belf.migrate.api.table.schema.TablePath;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.engine.job.*;
import belf.migrate.engine.job.structure.helper.CreateForeignKeyTool;
import belf.migrate.engine.job.structure.helper.CreateTableTool;
import belf.migrate.engine.job.structure.helper.DropTableTool;
import belf.migrate.api.taskconf.Task;
import belf.migrate.api.util.ExceptionUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.List;

import static belf.migrate.api.job.JobCode.*;

/**
 * 表结构同步的job执行类。该任务是批量执行，中间可能会因为执行失败而终止，可人工修复后继续执行任务。<br/>
 * 同步任务实现：在目标端创建表、主键、外键、索引。<br/>
 * 使用方式：
 * <ul>
 *     <li>创建表和外键，创建前先删除所有表：StructureJobRunner(job, true, true, true)</li>
 *     <li>创建表,不创建外键，创建前先删除所有表：StructureJobRunner(job, true, true, false)</li>
 *     <li>仅仅创建外键：StructureJobRunner(job, false, false, true)</li>
 *     <li>仅仅删除表：StructureJobRunner(job, false, true, false)</li>
 * </ul>
 */
@Slf4j
public class StructureJobRunner extends BaseJobRunner {
    private final Task task;
    private final TaskConf taskConf;
    /** 任务是否创建表？任务可以是仅仅创建外键，不创建表 */
    private boolean createTables = true;
    /** 是否创建表前先删除旧表 */
    private boolean dropTables = true;
    /** 是否创建外键？为了建表顺利且插入快，数据同步完成前可以先不创建外键 */
    private boolean createForeignKeys = false;

    public StructureJobRunner(Job job) {
        this(job, true, true, false);
    }

    public StructureJobRunner(Job job, boolean createTables, boolean dropTables, boolean createForeignKeys) {
        super(job);
        this.createTables = createTables;
        this.dropTables = dropTables;
        this.createForeignKeys = createForeignKeys;
        this.task = job.getTask();
        this.taskConf = task.getTaskConf();

    }

    @Override
    public JobContext getJobContext() {
        return jobContext;
    }


    @Override
    public JobResult check() {
        JobResult jobResult = new JobResult(SUCCESS, "任务执行成功");

        jobContext = new JobContext("" + job.getJobId(), taskConf);
        // 创建源端catalog
        AbstractJdbcCatalog sourceCatalog = CatalogCreator.createSourceCatalog(taskConf);
        if (null == sourceCatalog) {
            log.error("No Source Catalog for {}, Stop Job {}!",
                    taskConf.getSourceConf().getCatalogName(),
                    job.getJobId());

            jobResult = new JobResult(NO_SOURCE_CATALOG, "提交Job失败，没有找到目标端" +
                    taskConf.getSinkConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");

            return jobResult;
        }
        jobContext.setSourceCatalog(sourceCatalog);

        // 创建目标catalog
        AbstractSinkCatalog sinkCatalog = CatalogCreator.createSinkCatalog(taskConf);
        if (null == sinkCatalog) {
            log.error("No Sink Catalog for {}, Stop Job {}!",
                    taskConf.getSinkConf().getCatalogName(),
                    job.getJobId());

            jobResult = new JobResult(NO_SINK_CATALOG, "提交Job失败，没有找到源端" +
                    taskConf.getSourceConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");
            return jobResult;
        }
        jobContext.setSinkCatalog(sinkCatalog);

        try {
            // 打开源端数据库连接
            sourceCatalog.open();

            List<String> tableNames = job.getMissions();
            if (tableNames == null || tableNames.isEmpty()) {
                tableNames = sourceCatalog.listTables(jobContext.getTaskConf().getSourceConf().getDatabaseName(),
                        jobContext.getTaskConf().getSourceConf().getSchemaName());
            }
            String databaseName = jobContext.getTaskConf().getSourceConf().getDatabaseName();
            String schemaName = jobContext.getTaskConf().getSourceConf().getSchemaName();
            QualifiersModel qualifiersModel = jobContext.getSourceQualifiersModel();

            // 加载所有源端表元数据，缓存在jobContext中
            for (String tableName : tableNames) {
                String simpleTableName = tableName;
                if (tableName.contains(".")) {
                    simpleTableName = StringUtils.substringAfterLast(tableName, ".");
                }
                try {
                    CatalogTable catalogTable =
                            sourceCatalog.getTable(TablePath.of(databaseName, schemaName, simpleTableName, qualifiersModel));
                    jobContext.addSourceTable(jobContext.getTaskConf().getSourceConf().getCatalogName(),
                            jobContext.getFullTableName(tableName),
                            catalogTable);
                } catch (Exception e) {
                    log.error("无法获取表{}的信息", tableName);
                }
            }
        } catch (Exception e) {
            log.error("Got Source Table Metadata error: {}", e.getMessage());
            jobResult = new JobResult(GOT_SOURCE_METADATA_ERROR,
                    "获取源端表的Metadata失败！",
                    ExceptionUtils.toString(e));
            return jobResult;
        }


        try {
            // 打开目标端数据库连接
            sinkCatalog.open();

            List<String> tableNames = job.getMissions();
            if (tableNames == null || tableNames.isEmpty()) {
                tableNames = sourceCatalog.listTables(jobContext.getValidSourceDatabase(), jobContext.getValidSourceDatabase());
            }
            String databaseName = jobContext.getTaskConf().getSinkConf().getDatabaseName();
            String schemaName = jobContext.getTaskConf().getSinkConf().getSchemaName();
            QualifiersModel qualifiersModel = jobContext.getSinkQualifiersModel();

            // 加载所有目标端表元数据，缓存在jobContext中
            for (String tableName : tableNames) {
                String simpleTableName = tableName;
                if (tableName.contains(".")) {
                    simpleTableName = StringUtils.substringAfterLast(tableName, ".");
                }
                try {
                    CatalogTable catalogTable =
                            sinkCatalog.getTable(TablePath.of(databaseName, schemaName, simpleTableName, qualifiersModel));
                    jobContext.addSinkTable(jobContext.getTaskConf().getSinkConf().getCatalogName(),
                            jobContext.getFullTableName(tableName),
                            catalogTable);
                } catch (TableNotExistException | CatalogException e) {
//                    e.printStackTrace();
//                    log.error("Got Sink Table Metadata error: {}", e.getMessage());
/*                        jobResult = new JobResult(GOT_SINK_METADATA_ERROR,
                                "获取目标端表的Metadata失败！",
                                ExceptionUtils.toString(e));
                        return jobResult;*/
                }
            }
        } catch (Exception e) {
//            e.printStackTrace();
        }

        return jobResult;
    }

    @Override
    public JobResult execute() {
        JobResult jobResult = check();
//        if (jobResult.getCode() != SUCCESS) {
//            return jobResult;
//        }

        // 创建表前先删除旧表
        if (dropTables) {
            jobResult = new DropTableTool(jobContext, job).run();
            if (jobResult.getCode() != SUCCESS) {
                return jobResult;
            }
        }

        // 创建表、主键和索引
        if (createTables) {
            jobResult = new CreateTableTool(jobContext, job).run();
            if (jobResult.getCode() != SUCCESS) {
                return jobResult;
            }
        }

        // 创建外键
        if (createForeignKeys) {
            jobResult = new CreateForeignKeyTool(jobContext, job).run();
            if (jobResult.getCode() != SUCCESS) {
                return jobResult;
            }
        }

        return jobResult;
    }

    @Override
    public JobMode getMode() {
        return JobMode.BATCH;
    }

    @Override
    public Job getJob() {
        return job;
    }
}
